#include "ssdb_impl.h"
#include "leveldb/cache.h"
#include "leveldb/env.h"
#include "leveldb/filter_policy.h"
#include "leveldb/iterator.h"

#include "iterator.h"
#include "t_hash.h"
#include "t_kv.h"
#include "t_queue.h"
#include "t_zset.h"

namespace ssdb {

SSDBImpl::SSDBImpl() {
  ldb = NULL;
  binlogs = NULL;
}

SSDBImpl::~SSDBImpl() {
  if (binlogs) {
    binlogs = nullptr;
  }
  if (ldb) {
    delete ldb;
  }
  if (options.block_cache) {
    delete options.block_cache;
  }
  if (options.filter_policy) {
    delete options.filter_policy;
  }
}

std::shared_ptr<SSDB> SSDB::open(const Options &opt, const std::string &dir) {
  std::shared_ptr<SSDBImpl> ssdb = std::make_shared<SSDBImpl>();
  ssdb->options.create_if_missing = true;
  ssdb->options.max_open_files = opt.max_open_files;
  ssdb->options.filter_policy = leveldb::NewBloomFilterPolicy(10);
  ssdb->options.block_cache = leveldb::NewLRUCache(opt.cache_size * 1048576);
  ssdb->options.block_size = opt.block_size * 1024;
  ssdb->options.write_buffer_size = opt.write_buffer_size * 1024 * 1024;
  // ssdb->options.compaction_speed = opt.compaction_speed;
  if (opt.compression) {
    ssdb->options.compression = leveldb::kSnappyCompression;
  } else {
    ssdb->options.compression = leveldb::kNoCompression;
  }

  leveldb::Status status;

  status = leveldb::DB::Open(ssdb->options, dir, &ssdb->ldb);
  if (!status.ok()) {
    log_error("open db failed: %s", status.ToString().c_str());
    goto err;
  }
  ssdb->binlogs = std::make_unique<BinlogQueue>(ssdb->ldb, opt.binlog, opt.binlog_capacity);

  return ssdb;
err:
  return nullptr;
}

int SSDBImpl::flushdb() {
  Transaction trans(binlogs.get());
  int ret = 0;
  bool stop = false;
  while (!stop) {
    std::unique_ptr<leveldb::Iterator> it;
    leveldb::ReadOptions iterate_options;
    iterate_options.fill_cache = false;
    leveldb::WriteOptions write_opts;

    it.reset(ldb->NewIterator(iterate_options));
    it->SeekToFirst();
    for (int i = 0; i < 10000; i++) {
      if (!it->Valid()) {
        stop = true;
        break;
      }
      // log_debug("%s", hexmem(it->key().data(), it->key().size()).c_str());
      leveldb::Status s = ldb->Delete(write_opts, it->key());
      if (!s.ok()) {
        log_error("del error: %s", s.ToString().c_str());
        stop = true;
        ret = -1;
        break;
      }
      it->Next();
    }
  }
  binlogs->flush();
  return ret;
}

std::shared_ptr<Iterator> SSDBImpl::iterator(const std::string &start, const std::string &end,
                             uint64_t limit) {
  std::shared_ptr<leveldb::Iterator> it;
  leveldb::ReadOptions iterate_options;
  iterate_options.fill_cache = false;
  it.reset(ldb->NewIterator(iterate_options));
  it->Seek(start);
  if (it->Valid() && it->key() == start) {
    it->Next();
  }
  return std::make_shared<Iterator>(it, end, limit);
}

std::shared_ptr<Iterator> SSDBImpl::rev_iterator(const std::string &start,
                                 const std::string &end, uint64_t limit) {
  std::shared_ptr<leveldb::Iterator> it;
  leveldb::ReadOptions iterate_options;
  iterate_options.fill_cache = false;
  it.reset(ldb->NewIterator(iterate_options));
  it->Seek(start);
  if (!it->Valid()) {
    it->SeekToLast();
  } else {
    it->Prev();
  }
  return std::make_shared<Iterator>(it, end, limit, Iterator::BACKWARD);
}

/* raw operates */

int SSDBImpl::raw_set(const Bytes &key, const Bytes &val) {
  leveldb::WriteOptions write_opts;
  leveldb::Status s = ldb->Put(write_opts, slice(key), slice(val));
  if (!s.ok()) {
    log_error("set error: %s", s.ToString().c_str());
    return -1;
  }
  return 1;
}

int SSDBImpl::raw_del(const Bytes &key) {
  leveldb::WriteOptions write_opts;
  leveldb::Status s = ldb->Delete(write_opts, slice(key));
  if (!s.ok()) {
    log_error("del error: %s", s.ToString().c_str());
    return -1;
  }
  return 1;
}

int SSDBImpl::raw_get(const Bytes &key, std::string *val) {
  leveldb::ReadOptions opts;
  opts.fill_cache = false;
  leveldb::Status s = ldb->Get(opts, slice(key), val);
  if (s.IsNotFound()) {
    return 0;
  }
  if (!s.ok()) {
    log_error("get error: %s", s.ToString().c_str());
    return -1;
  }
  return 1;
}

uint64_t SSDBImpl::size() {
  std::string s = "A";
  std::string e(1, 'z' + 1);
  leveldb::Range ranges[1];
  ranges[0] = leveldb::Range(s, e);
  uint64_t sizes[1];
  ldb->GetApproximateSizes(ranges, 1, sizes);
  return sizes[0];
}

std::vector<std::string> SSDBImpl::info() {
  //  "leveldb.num-files-at-level<N>" - return the number of files at level <N>,
  //     where <N> is an ASCII representation of a level number (e.g. "0").
  //  "leveldb.stats" - returns a multi-line string that describes statistics
  //     about the internal operation of the DB.
  //  "leveldb.sstables" - returns a multi-line string that describes all
  //     of the sstables that make up the db contents.
  std::vector<std::string> info;
  std::vector<std::string> keys;
  /*
  for(int i=0; i<7; i++){
          char buf[128];
          snprintf(buf, sizeof(buf), "leveldb.num-files-at-level%d", i);
          keys.push_back(buf);
  }
  */
  keys.push_back("leveldb.stats");
  // keys.push_back("leveldb.sstables");

  for (size_t i = 0; i < keys.size(); i++) {
    std::string key = keys[i];
    std::string val;
    if (ldb->GetProperty(key, &val)) {
      info.push_back(key);
      info.push_back(val);
    }
  }

  return info;
}

void SSDBImpl::compact() { ldb->CompactRange(NULL, NULL); }

int SSDBImpl::key_range(std::vector<std::string> *keys) {
  int ret = 0;
  std::string kstart, kend;
  std::string hstart, hend;
  std::string zstart, zend;
  std::string qstart, qend;


  auto it = this->iterator(encode_kv_key(""), "", 1);
  if (it->next()) {
    Bytes ks = it->key();
    if (ks.data()[0] == DataType::KV) {
      std::string n;
      if (decode_kv_key(ks, &n) == -1) {
        ret = -1;
      } else {
        kstart = n;
      }
    }
  }

  it = this->rev_iterator(encode_kv_key("\xff"), "", 1);
  if (it->next()) {
    Bytes ks = it->key();
    if (ks.data()[0] == DataType::KV) {
      std::string n;
      if (decode_kv_key(ks, &n) == -1) {
        ret = -1;
      } else {
        kend = n;
      }
    }
  }

  it = this->iterator(encode_hsize_key(""), "", 1);
  if (it->next()) {
    Bytes ks = it->key();
    if (ks.data()[0] == DataType::HSIZE) {
      std::string n;
      if (decode_hsize_key(ks, &n) == -1) {
        ret = -1;
      } else {
        hstart = n;
      }
    }
  }

  it = this->rev_iterator(encode_hsize_key("\xff"), "", 1);
  if (it->next()) {
    Bytes ks = it->key();
    if (ks.data()[0] == DataType::HSIZE) {
      std::string n;
      if (decode_hsize_key(ks, &n) == -1) {
        ret = -1;
      } else {
        hend = n;
      }
    }
  }

  it = this->iterator(encode_zsize_key(""), "", 1);
  if (it->next()) {
    Bytes ks = it->key();
    if (ks.data()[0] == DataType::ZSIZE) {
      std::string n;
      if (decode_zsize_key(ks, &n) == -1) {
        ret = -1;
      } else {
        zstart = n;
      }
    }
  }

  it = this->rev_iterator(encode_zsize_key("\xff"), "", 1);
  if (it->next()) {
    Bytes ks = it->key();
    if (ks.data()[0] == DataType::ZSIZE) {
      std::string n;
      if (decode_zsize_key(ks, &n) == -1) {
        ret = -1;
      } else {
        zend = n;
      }
    }
  }

  it = this->iterator(encode_qsize_key(""), "", 1);
  if (it->next()) {
    Bytes ks = it->key();
    if (ks.data()[0] == DataType::QSIZE) {
      std::string n;
      if (decode_qsize_key(ks, &n) == -1) {
        ret = -1;
      } else {
        qstart = n;
      }
    }
  }

  it = this->rev_iterator(encode_qsize_key("\xff"), "", 1);
  if (it->next()) {
    Bytes ks = it->key();
    if (ks.data()[0] == DataType::QSIZE) {
      std::string n;
      if (decode_qsize_key(ks, &n) == -1) {
        ret = -1;
      } else {
        qend = n;
      }
    }
  }

  keys->push_back(kstart);
  keys->push_back(kend);
  keys->push_back(hstart);
  keys->push_back(hend);
  keys->push_back(zstart);
  keys->push_back(zend);
  keys->push_back(qstart);
  keys->push_back(qend);

  return ret;
}
}
